package com.uestc.coprocessor.registerLocation;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import com.uestc.coprocessor.registerLocation.RegisterLocationProtocol;
import com.uestc.coprocessor.registerLocation.RegisterLocationProtocol.CountRequest;
import com.uestc.coprocessor.registerLocation.RegisterLocationProtocol.CountResponse;

public class ProvinceRegisterStatistics extends RegisterLocationProtocol.ProvinceRegisterStatistic
		implements Coprocessor, CoprocessorService {
	private RegionCoprocessorEnvironment env;

	/**
	 * 统计各个省份的注册人数
	 */
	@Override
	public void getUsersInProvinces(RpcController controller, CountRequest request, RpcCallback<CountResponse> done) {
		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("province"));
		RegisterLocationProtocol.CountResponse response = null;
		InternalScanner scanner = null;
		LinkedHashMap<String, Long> map = (LinkedHashMap<String, Long>) ProvinceInfo.getMap();
		try {
			scanner = env.getRegion().getScanner(scan);
			List<Cell> results = new ArrayList<Cell>();
			boolean hasMore = false;
			do {
				hasMore = scanner.next(results);
				for (Cell kv : results) {
//					String info = Bytes.toString(CellUtil.cloneFamily(kv));
//					String qualifier = Bytes.toString(CellUtil.cloneValue(kv));
					String province = Bytes.toString(CellUtil.cloneValue(kv)).toLowerCase();
//					if (!"info".equals(info) || !"province".equals(qualifier) || province == null || province.isEmpty())
//						continue;
					Long value = map.get(province);
					if (value == null) { // 不在中国境内，暂时不统计
						continue;
					} else {
						map.put(province, ++value);
					}
				}
				results.clear();
			} while (hasMore);

			// 生成返回给调用端的字符串
			StringBuilder sb = new StringBuilder();
			// 按顺序遍历一次Map，进行两两相加
			int i = 0;
			String province = null;
			Long value = 0l;
			for (Map.Entry<String, Long> item : map.entrySet()) {
				if (i % 2 == 0) {
					value = item.getValue();
					province = item.getKey();
				} else {
					sb.append(String.format("%s:%s;", province, value + item.getValue()));
				}
				++i;
			}

			// 生成response, result 是各省的串
			response = RegisterLocationProtocol.CountResponse.newBuilder().setResult(sb.toString()).build();

		} catch (IOException e) {
			e.printStackTrace();
			ResponseConverter.setControllerException(controller, e);
		} finally {
			if (scanner != null) {
				try {
					scanner.close();
				} catch (IOException ignored) {
				}
			}
		}
		done.run(response);

	}

	@Override
	public Service getService() {
		return this;
	}

	/**
	 * 在HBase启动或加载Coprocessor的时候会调用
	 */
	@Override
	public void start(CoprocessorEnvironment arg0) throws IOException {
		if (arg0 instanceof RegionCoprocessorEnvironment) {
			this.env = (RegionCoprocessorEnvironment) arg0;
		} else {
			throw new CoprocessorException("Must be loaded on a table region!");
		}

	}

	@Override
	public void stop(CoprocessorEnvironment arg0) throws IOException {
	}

}
